package com.xiaohu.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo17Checkpoint {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("缓存演示")
    val sc: SparkContext = new SparkContext(conf)
    //设置检查点的存储路径
    sc.setCheckpointDir("spark/data/checkpoint1")
    //===================================================================

    val linesRDD: RDD[String] = sc.textFile("spark/data/students.txt")
    val studentsRDD: RDD[Student2] = linesRDD.map(_.split(","))
      .map {
        case Array(id: String, name: String, age: String, gender: String, clazz: String) =>
          Student2(id, name, age.toInt, gender, clazz)
      }

    /**
     * 永久将执行过程中RDD中流动的数据存储到磁盘（hdfs）中
     * checkpoint
     *
     * 需要设置checkpoint的路径，统一设置的
     *
     * checkpoint也相当于一个行动算子，触发作业执行
     * 第二次DAG有向无环图执行的时候，直接从最后一个有检查点的rdd开始向下执行
     */
    studentsRDD.checkpoint()



    //需求1：求每个班级的人数
    val rdd1: RDD[(String, Iterable[Student2])] = studentsRDD.groupBy(_.clazz)
    val resRDD1: RDD[(String, Int)] = rdd1.map((kv: (String, Iterable[Student2])) => (kv._1, kv._2.size))
    resRDD1.foreach(println)

    //需求2：求每个年龄的人数
    val rdd2: RDD[(Int, Iterable[Student2])] = studentsRDD.groupBy(_.age)
    val resRDD2: RDD[(Int, Int)] = rdd2.map((kv: (Int, Iterable[Student2])) => (kv._1, kv._2.size))
    resRDD2.foreach(println)

    while (true) {

    }
  }
}

/**
 * checkpoint和cache的区别？
 * cache是将一个复杂的RDD做缓存，将来执行的时候，只是这个rdd会从缓存中取
 * checkpoint是永久将rdd数据持久化，将来执行的时候，直接从检查点的rdd往后执行
 */
